package worker

import (
	"encoding/json"
	"fmt"
	"github.com/avast/retry-go"
	"time"
	"wf-deploy-agent/cmd"
	"wf-deploy-agent/cmd/issue"
	"wf-deploy-agent/cmd/issue/base"
	"wf-deploy-agent/conf"
	"wf-deploy-agent/infra"
	"wf-deploy-agent/notify"
	"wf-deploy-agent/remote/entity"
	"wf-deploy-agent/util"
)

// Issue 文件分发处理程序
type Issue struct {
}

func (d *Issue) process() {
	err := notify.Subscribe(conf.CF.Kafka.IssueTopic, fmt.Sprintf("agent-%s:%d",
		util.GetLocalIp(),
		conf.CF.Agent.Service.Port),
		d.handle, d.post)
	if err != nil {
		infra.Logger.Panic(err)
	}
}

func (d *Issue) handle(command *cmd.Command) error {
	issueCommand := &base.IssueCommand{}
	err := json.Unmarshal([]byte(command.Raw), issueCommand)
	if err != nil {
		return err
	}

	ts := time.UnixMilli(issueCommand.Timestamp)
	span := time.Now().Sub(ts)
	if span > 10*time.Minute {
		return nil
	}

	journal := &entity.IssueNodeJournal{
		OrderId:    issueCommand.OrderId,
		BatchNO:    command.BatchNO,
		ResourceId: conf.CF.Agent.Id,
		State:      1,
	}

	err = d.notify(journal)
	if err != nil {
		return err
	}

	issueCommand.Raw = command.Raw

	entrance := &issue.Entrance{}
	return entrance.Exec(issueCommand)
}

func (d *Issue) notify(journal *entity.IssueNodeJournal) error {
	return retry.Do(func() error {
		return notify.Broadcast(conf.CF.Kafka.IssueNodeTopic, journal)
	})
}

func (d *Issue) post(command *cmd.Command, err error) error {
	issueCommand := &base.IssueCommand{}
	err0 := json.Unmarshal([]byte(command.Raw), issueCommand)
	if err0 != nil {
		return err0
	}

	msg := ""
	state := 3
	if err != nil {
		state = 2
		msg = fmt.Sprint(err)
	}

	journal := &entity.IssueNodeJournal{
		OrderId:    issueCommand.OrderId,
		BatchNO:    command.BatchNO,
		ResourceId: conf.CF.Agent.Id,
		State:      state,
		Result:     msg,
		FinishTime: time.Now().UnixMilli(),
	}

	return d.notify(journal)
}
